package com.nanohadoop.hdfs;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.io.*;
import java.net.*;
import com.nanohadoop.utils.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * NameNode 作为 HDFS 的管理节点,主要负责:
 * 1. 管理文件系统命名空间
 * 2. 管理数据块映射信息 
 * 3. 处理客户端的读写请求
 */
public class NameNode implements Serializable {
    private static final Logger log = LoggerFactory.getLogger(NameNode.class);
    // 文件系统目录树
    private final FSDirectory fsDirectory;
    
    // 数据块到DataNode的映射
    private final BlocksMap blocksMap;
    
    // 活跃DataNode列表
    private final Map<String, DataNodeInfo> datanodes;

    // 配置信息
    private final Configuration conf;
    
    // 副本数量
    private final int replicationFactor;
    
    // 数据块大小
    private final long blockSize;
    
    // 服务器线程
    private Thread serverThread;
    private ServerSocket serverSocket;
    private volatile boolean running;

    public NameNode() {
        this(Configuration.getInstance());
    }
    
    public NameNode(Configuration conf) {
        this.conf = conf;
        this.fsDirectory = new FSDirectory();
        this.blocksMap = new BlocksMap();
        this.datanodes = new ConcurrentHashMap<>();
        
        // 从配置文件加载参数
        this.replicationFactor = conf.getInt("dfs.replication", 3);
        this.blockSize = conf.getLong("dfs.block.size", 64 * 1024 * 1024); // 默认64MB
        
        // 加载namenode目录
        String nameDirPath = conf.get("dfs.namenode.name.dir", "file:///tmp/dfs/name");
        nameDirPath = nameDirPath.replace("file://", "");
        initializeNameNodeDirectory(nameDirPath);
    }
    
    private void initializeNameNodeDirectory(String dirPath) {
        try {
            java.io.File nameDir = new java.io.File(dirPath);
            if (!nameDir.exists()) {
                nameDir.mkdirs();
            }
        } catch (Exception e) {
            throw new RuntimeException("Failed to initialize NameNode directory: " + dirPath, e);
        }
    }

    public synchronized void start() throws IOException {
        if (running) {
            return;
        }
        running = true;
        
        // 获取配置的端口号
        int port = conf.getInt("dfs.namenode.port", 9000);
        serverSocket = new ServerSocket(port);
        
        // 启动服务线程
        serverThread = new Thread(() -> {
            while (running) {
                try {
                    Socket socket = serverSocket.accept();
                    // 处理客户端请求
                    new Thread(() -> handleClientRequest(socket)).start();
                } catch (IOException e) {
                    if (running) {
                        e.printStackTrace();
                    }
                }
            }
        });
        serverThread.start();
    }

    public synchronized void stop() {
        if (!running) {
            return;
        }
        running = false;
        try {
            if (serverSocket != null) {
                serverSocket.close();
            }
            if (serverThread != null) {
                serverThread.join();
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void handleClientRequest(Socket socket) {
        try (DataInputStream in = new DataInputStream(socket.getInputStream());
             DataOutputStream out = new DataOutputStream(socket.getOutputStream())) {
            
            String command = in.readUTF();
            
            switch (command) {
                case "REGISTER_DATANODE":
                    String datanodeId = in.readUTF();
                    String host = in.readUTF();
                    int port = in.readInt();
                    registerDataNode(datanodeId, host, port);
                    out.writeBoolean(true);
                    break;
                    
                case "CREATE":
                    String path = in.readUTF();
                    boolean result = createFile(path);
                    out.writeBoolean(result);
                    break;
                    
                case "EXISTS":
                    path = in.readUTF();
                    result = fsDirectory.exists(path);
                    out.writeBoolean(result);
                    break;
                    
                case "DELETE":
                    path = in.readUTF();
                    result = fsDirectory.delete(path);
                    out.writeBoolean(result);
                    break;
                    
                case "GET_BLOCK_LOCATIONS":
                    path = in.readUTF();
                    long blockId = in.readLong();
                    long blockLength = in.readLong();  // 接收块大小
                    List<DataNodeInfo> locations = allocateBlock(path, blockId, blockLength);
                    out.writeInt(locations.size());
                    for (DataNodeInfo node : locations) {
                        out.writeUTF(node.getId());
                        out.writeUTF(node.getHost());
                        out.writeInt(node.getPort());
                    }
                    break;
                    
                case "GET_FILE_STATUS":
                    path = in.readUTF();
                    // 处理获取文件状态的请求
                    if (!fsDirectory.exists(path)) {
                        out.writeBoolean(false);
                    } else {
                        out.writeBoolean(true);
                        INode inode = fsDirectory.getINode(path);
                        out.writeLong(inode.getLength());  // length
                        out.writeBoolean(inode.isDirectory);  // isDirectory
                        out.writeInt(replicationFactor);  // replication
                        out.writeLong(blockSize);  // blockSize
                    }
                    break;
                    
                case "GET_FILE_BLOCKS":
                    path = in.readUTF();
                    log.debug("GET_FILE_BLOCKS request for: " + path);
                    if (!fsDirectory.exists(path)) {
                        log.debug("File does not exist: " + path);
                        out.writeInt(0);  // 没有块
                    } else {
                        INode inode = fsDirectory.getINode(path);
                        List<Long> blocks = inode.blocks;
                        log.debug("Found " + blocks.size() + " blocks for file: " + path);
                        
                        // 写出块数量
                        out.writeInt(blocks.size());
                        
                        // 对每个块
                        for (Long bid : blocks) {
                            out.writeLong(bid);  // 写出块ID
                            
                            // 获取块的位置信息
                            List<DataNodeInfo> blockLocations = blocksMap.getLocations(bid);
                            log.debug("Block " + bid + " has " + blockLocations.size() + " locations");
                            
                            // 写出位置数量和详细信息
                            out.writeInt(blockLocations.size());
                            for (DataNodeInfo node : blockLocations) {
                                out.writeUTF(node.getId());
                                out.writeUTF(node.getHost());
                                out.writeInt(node.getPort());
                            }
                        }
                        log.debug("Successfully sent file blocks info");
                    }
                    break;
                    
                case "HEARTBEAT":
                    datanodeId = in.readUTF();
                    long totalSpace = in.readLong();
                    long usedSpace = in.readLong();
                    DataNodeInfo dataNode = getDataNodeInfo(datanodeId);
                    if (dataNode != null) {
                        dataNode.updateHeartbeat();
                        dataNode.updateStorageInfo(totalSpace, usedSpace);
                        int numBlocks = in.readInt();
                        List<Long> blockIds = new ArrayList<>();
                        for (int i = 0; i < numBlocks; i++) {
                            blockIds.add(in.readLong());
                        }
                        handleHeartbeat(datanodeId, blockIds);
                    }
                    out.writeBoolean(true);
                    break;
                    
                default:
                    out.writeBoolean(false);
            }
            
            out.flush();
            out.flush();
            
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    // 处理客户端创建文件请求
    public synchronized boolean createFile(String path) {
        log.debug("Creating file: " + path);
        boolean result = fsDirectory.createFile(path);
        log.debug("File creation " + (result ? "succeeded" : "failed") + ": " + path);
        return result;
    }

    // 处理客户端写入数据块请求
    public List<DataNodeInfo> allocateBlock(String filename, long blockId, long blockLength) {
        log.debug("Allocating block " + blockId + " for file " + filename);
        
        // 根据配置的副本数选择DataNode
        List<DataNodeInfo> locations = chooseDataNodes(replicationFactor);
        if (locations.isEmpty()) {
            log.debug("WARNING: No DataNodes available for block allocation");
            return locations;
        }
        
        // 添加块映射并更新文件长度
        blocksMap.addBlock(blockId, locations);
        INode inode = fsDirectory.getINode(filename);
        if (inode != null) {
            inode.addBlock(blockId);
            inode.setLength(inode.getLength() + blockLength);
            log.debug("File " + filename + ": added block " + blockId + 
                ", new length=" + inode.getLength() + ", block count=" + inode.blocks.size());
        } else {
            log.debug("ERROR: INode not found for file " + filename);
        }
        
        log.debug("Allocated block " + blockId + " to DataNodes: " + 
            locations.stream()
                    .map(node -> node.getHost() + ":" + node.getPort())
                    .reduce((a, b) -> a + ", " + b)
                    .orElse(""));
        
        return locations;
    }

    // 处理DataNode的心跳
    public synchronized void handleHeartbeat(String datanodeId, List<Long> blocks) {
        DataNodeInfo node = datanodes.get(datanodeId);
        if (node != null) {
            node.updateHeartbeat();
            blocksMap.updateBlocks(datanodeId, blocks);
        }
    }

    // 获取块大小
    public long getBlockSize() {
        return blockSize;
    }

    // 注册新的DataNode
    public synchronized void registerDataNode(String datanodeId, String host, int port) {
        if (!datanodes.containsKey(datanodeId)) {
            datanodes.put(datanodeId, new DataNodeInfo(datanodeId, host, port));
            log.debug("DataNode registered: " + datanodeId);
        }
    }

    // 获取DataNode信息
    public DataNodeInfo getDataNodeInfo(String datanodeId) {
        return datanodes.get(datanodeId);
    }

    // 获取所有活跃的DataNode
    public List<DataNodeInfo> getActiveDataNodes() {
        List<DataNodeInfo> active = datanodes.values().stream()
                .filter(DataNodeInfo::isAlive)
                .collect(java.util.stream.Collectors.toList());
        log.debug("Active DataNodes: " + active.size());
        return active;
    }

    // 选择指定数量的DataNode用于存储副本
    private List<DataNodeInfo> chooseDataNodes(int numReplicas) {
        List<DataNodeInfo> chosen = new ArrayList<>();
        List<DataNodeInfo> active = getActiveDataNodes();
        if (active.isEmpty()) {
            log.debug("No active DataNodes found!");
            return chosen;
        }
        // 简单实现:随机选择
        for(int i = 0; i < numReplicas && i < active.size(); i++) {
            chosen.add(active.get(i));
        }
        log.debug("Chosen DataNodes: " + chosen.size());
        return chosen;
    }

    /**
     * DataNodeInfo 是 NameNode 用于管理 DataNode 的元数据信息类
     * 它记录了 DataNode 的状态信息，而不是实际的 DataNode 服务器实现
     */
    public static class DataNodeInfo implements Serializable {
        private final String id;           // DataNode 的唯一标识
        private final String host;         // DataNode 主机地址
        private final int port;            // DataNode 服务端口
        private long lastHeartbeat;        // 最后一次心跳时间
        private long totalSpace;           // 总存储空间
        private long usedSpace;            // 已使用空间
        private List<Long> blocks;         // 当前存储的数据块列表
        private boolean isAlive;           // 节点是否存活
        
        public DataNodeInfo(String id, String host, int port) {
            this.id = id;
            this.host = host;
            this.port = port;
            this.lastHeartbeat = System.currentTimeMillis();
            this.blocks = new ArrayList<>();
            this.isAlive = true;
            this.totalSpace = 0;
            this.usedSpace = 0;
        }
        
        public String getId() {
            return id;
        }

        public String getHost() {
            return host;
        }

        public int getPort() {
            return port;
        }
        
        public long getLastHeartbeat() {
            return lastHeartbeat;
        }

        public boolean isAlive() {
            return isAlive;
        }

        public List<Long> getBlocks() {
            return new ArrayList<>(blocks);
        }

        public double getUsageRatio() {
            return totalSpace == 0 ? 0 : (double) usedSpace / totalSpace;
        }
        
        public void updateHeartbeat() {
            this.lastHeartbeat = System.currentTimeMillis();
            this.isAlive = true;
        }

        public void markDead() {
            this.isAlive = false;
        }

        public void updateStorageInfo(long totalSpace, long usedSpace) {
            this.totalSpace = totalSpace;
            this.usedSpace = usedSpace;
        }

        public void updateBlocks(List<Long> blocks) {
            this.blocks = new ArrayList<>(blocks);
        }

        @Override
        public String toString() {
            return String.format("DataNodeInfo{id=%s, host=%s, port=%d, alive=%s, usage=%.2f%%}", 
                id, host, port, isAlive, getUsageRatio() * 100);
        }
    }

    /**
     * 表示数据块的位置信息
     */
    public static class BlockLocation implements Serializable {
        private final long blockId;
        private final List<DataNodeInfo> locations;
        
        public BlockLocation(long blockId, List<DataNodeInfo> locations) {
            this.blockId = blockId;
            this.locations = locations;
        }
        
        public long getBlockId() {
            return blockId;
        }
        
        public List<DataNodeInfo> getLocations() {
            return locations;
        }
    }
}

// 文件系统目录树
class FSDirectory {
    private final Map<String, INode> inodes = new HashMap<>();
    
    public boolean createFile(String path) {
        if(inodes.containsKey(path)) {
            return false;
        }
        inodes.put(path, new INode(path, false));
        return true;
    }
    
    public boolean exists(String path) {
        return inodes.containsKey(path);
    }
    
    public boolean delete(String path) {
        return inodes.remove(path) != null;
    }

    public INode getINode(String path) {
        return inodes.get(path);
    }
}

// 数据块映射管理
class BlocksMap {
    private static final Logger log = LoggerFactory.getLogger(BlocksMap.class);
    private final Map<Long, List<NameNode.DataNodeInfo>> blockToNodes = new HashMap<>();
    
    public void addBlock(long blockId, List<NameNode.DataNodeInfo> locations) {
        blockToNodes.put(blockId, new ArrayList<>(locations));
    }
    
    public void updateBlocks(String datanodeId, List<Long> blocks) {
        // 先从所有块中移除该DataNode
        for (List<NameNode.DataNodeInfo> nodeList : blockToNodes.values()) {
            nodeList.removeIf(node -> node.getId().equals(datanodeId));
        }
        
        // 更新DataNode的块列表
        NameNode.DataNodeInfo nodeInfo = null;
        for (List<NameNode.DataNodeInfo> nodeList : blockToNodes.values()) {
            for (NameNode.DataNodeInfo node : nodeList) {
                if (node.getId().equals(datanodeId)) {
                    nodeInfo = node;
                    break;
                }
            }
            if (nodeInfo != null) break;
        }
        
        if (nodeInfo != null) {
            nodeInfo.updateBlocks(blocks);
        }
    }
    
    public List<NameNode.DataNodeInfo> getLocations(long blockId) {
        List<NameNode.DataNodeInfo> locations = blockToNodes.get(blockId);
        if (locations == null) {
            log.debug("Warning: No locations found for block " + blockId);
            return new ArrayList<>();
        }
        // 过滤掉不活跃的DataNode
        List<NameNode.DataNodeInfo> activeLocations = new ArrayList<>();
        for (NameNode.DataNodeInfo node : locations) {
            if (node.isAlive()) {
                activeLocations.add(node);
            }
        }
        if (activeLocations.isEmpty()) {
            log.debug("Warning: No active locations found for block " + blockId);
        }
        return activeLocations;
    }
}

// 文件系统树节点
class INode {
    String path;
    boolean isDirectory;
    List<Long> blocks;
    long length;  // 文件长度
    
    public INode(String path, boolean isDirectory) {
        this.path = path;
        this.isDirectory = isDirectory;
        this.blocks = new ArrayList<>();
        this.length = 0;
    }
    
    public void addBlock(long blockId) {
        blocks.add(blockId);
    }
    
    public void setLength(long length) {
        this.length = length;
    }
    
    public long getLength() {
        return length;
    }
}